package com.shuyun.service;

import com.alibaba.fastjson.JSON;
import com.aliyun.ons20190214.models.OnsGroupCreateRequest;
import com.aliyun.ons20190214.models.OnsGroupCreateResponse;
import com.aliyun.tea.TeaException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Map;

/**
 * @description:
 * @author: shuai.chen@shuyun.com
 * @date: 2022/10/9 3:16 PM
 **/
@Service
@Slf4j
public class RocketMQService {

    /**
     * RocketMQTemplate 默认使用
     *     private DefaultMQProducer producer;
     *     private DefaultLitePullConsumer consumer;
     */
    @Autowired
    private Map<String, RocketMQTemplate> rocketMQTemplateMap;


    @Autowired
    private DefaultMQProducer defaultMQProducer;

    /**
     * 配置信息都注入这里了
     */
    @Autowired
    private RocketMQProperties rocketMQProperties;

    @Autowired
    private com.aliyun.ons20190214.Client onsClient;

    /**
     * 根据标签获取RocketMQTemplate
     *
     * @param tags
     *            标签
     * @return RocketMQTemplate
     */
    public RocketMQTemplate getRocketMQTemplateByTag(String tags) {
        RocketMQTemplate rocketMQTemplate = rocketMQTemplateMap.get(tags);
        if (null == rocketMQTemplate) {
            rocketMQTemplate = createRocketMQTemplateByTag(tags);
            rocketMQTemplateMap.put(tags, rocketMQTemplate);
        }
        return rocketMQTemplate;
    }

    /**
     * 根据标签创建RocketMQTemplate
     * 
     * @param tags
     *            标签
     * @return RocketMQTemplate
     */
    private RocketMQTemplate createRocketMQTemplateByTag(String tags) {
        RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
        rocketMQTemplate.setProducer(defaultMQProducer);
        DefaultLitePullConsumer consumer = createRocketMQConsumerByTag(tags);
        try {
            consumer.start();
        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }
        rocketMQTemplate.setConsumer(consumer);
        return rocketMQTemplate;
    }

    /**
     * 根据标签创建RocketMQ消费者
     * 
     * @param tags
     *            标签
     * @return 消费者
     */
    private DefaultLitePullConsumer createRocketMQConsumerByTag(String tags) {
        RocketMQProperties.Consumer consumerConfig = rocketMQProperties.getConsumer();
        String nameServer = rocketMQProperties.getNameServer();
        String namespace = consumerConfig.getNamespace();
        String groupName = getGroupName(namespace, tags);
        String topicName = consumerConfig.getTopic();
        String accessChannel = rocketMQProperties.getAccessChannel();
        MessageModel messageModel = MessageModel.valueOf(consumerConfig.getMessageModel());
        SelectorType selectorType = SelectorType.valueOf(consumerConfig.getSelectorType());
        String selectorExpression = tags;
        String ak = consumerConfig.getAccessKey();
        String sk = consumerConfig.getSecretKey();
        int pullBatchSize = consumerConfig.getPullBatchSize();
        boolean useTLS = consumerConfig.isTlsEnable();

        DefaultLitePullConsumer litePullConsumer = null;
        try {
            litePullConsumer = RocketMQUtil.createDefaultLitePullConsumer(nameServer, accessChannel, groupName,
                topicName, messageModel, selectorType, selectorExpression, ak, sk, pullBatchSize, useTLS);
        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }
        litePullConsumer.setEnableMsgTrace(consumerConfig.isEnableMsgTrace());
        litePullConsumer.setCustomizedTraceTopic(consumerConfig.getCustomizedTraceTopic());
        litePullConsumer.setNamespace(namespace);
        return litePullConsumer;
    }

    /**
     * 根据标签获取消费组
     * 
     * @param tags
     *            标签
     * @return 消费组
     */
    private String getGroupName(String namespace, String tags) {
        String groupId = "GID_" + tags;
        OnsGroupCreateRequest onsGroupCreateRequest =
            new OnsGroupCreateRequest().setGroupId(groupId).setInstanceId(namespace).setGroupType("tcp");
        log.info("OnsGroupCreateRequest:{}", JSON.toJSONString(onsGroupCreateRequest));
        try {
            // 复制代码运行请自行打印 API 的返回值
            OnsGroupCreateResponse response = onsClient.onsGroupCreate(onsGroupCreateRequest);
            log.info("getGroupName:{}", JSON.toJSONString(response));
            return groupId;
        } catch (TeaException error) {
            // 如有需要，请打印 error
            com.aliyun.teautil.Common.assertAsString(error.message);
            if(StringUtils.contains(error.message,"specified consumerId has already exist")){
                return groupId;
            }else{
                throw error;
            }
        } catch (Exception _error) {
            TeaException error = new TeaException(_error.getMessage(), _error);
            // 如有需要，请打印 error
            com.aliyun.teautil.Common.assertAsString(error.message);
            throw error;
        }
    }
}
